package cn.getech.data.metadata.service.impl;

import cn.getech.data.development.constant.StoreTableType;
import cn.getech.data.development.constant.TableRegisterTypeEnum;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.entity.flink.BdpRealTableField;
import cn.getech.data.development.entity.flink.BdpRealTableInfo;
import cn.getech.data.development.mapper.flink.BdpRealTableFieldMapper;
import cn.getech.data.development.service.*;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.metadata.constant.TaskStatusEnum;
import cn.getech.data.metadata.constant.TaskTypeEnum;
import cn.getech.data.metadata.entity.meta.CollectionTableField;
import cn.getech.data.metadata.entity.meta.CollectionTableInfo;
import cn.getech.data.metadata.entity.meta.CollectionTaskLog;
import cn.getech.data.metadata.mapper.BloodRelationshipMapper;
import cn.getech.data.metadata.model.dto.CollectionSourceDto;
import cn.getech.data.metadata.model.res.CollectionSystemModuleListRes;
import cn.getech.data.metadata.model.res.CollectionSystemModuleRes;
import cn.getech.data.metadata.service.*;
import cn.getech.data.metadata.service.CollectionTableInfoService;
import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

@Service("collectionRegisterService")
public class CollectionRegisterServiceImpl implements CollectionRegisterService {
    private Logger log = LoggerFactory.getLogger(CollectionRegisterServiceImpl.class);

    @Autowired
    private TableInfoService tableInfoService;
    @Autowired
    private TableFieldInfoService tableFieldInfoService;

    @Autowired
    private BdpRealTableInfoService realTableInfoService;

    @Autowired
    private CollectionTableInfoService collectionTableInfoService;
    @Autowired
    private CollectionTableField2Service collectionTableField2Service;

    @Autowired
    private TableRegisterService tableRegisterService;
    @Autowired
    private TableRegisterHistoryService tableRegisterHistoryService;
    @Autowired
    private CollectionTaskService collectionTaskService;
    @Autowired
    private CollectionGenerateService collectionGenerateService;
    @Autowired
    private BloodRelationshipMapper bloodRelationshipMapper;
    @Override
    public void init() {
        Date runTime = new Date();
        List<TableInfo> offList = tableInfoService.list(new QueryWrapper<>());
        List<BdpRealTableInfo> onlineList = realTableInfoService.list(new QueryWrapper<BdpRealTableInfo>()
                .eq("delete_flag", 0));
        List<Integer> offLineTableIds = offList.stream().filter(s -> s.getId() != null)
                .map(TableInfo::getId).collect(Collectors.toList());
        List<Integer> onLineTableIds = onlineList.stream().filter(s -> s.getId() != null)
                .map(BdpRealTableInfo::getId).collect(Collectors.toList());
        List<CollectionTableInfo> sourceTables = transform(offLineTableIds, onLineTableIds);

        List<CollectionTableInfo> targetTables = collectionTableInfoService.list(
                new QueryWrapper<CollectionTableInfo>()
                        .eq("delete_flag", 0)
                        .in("table_type",Arrays.asList(1,2))
        );
        Function<String, Long> insertRunLog = new Function<String, Long>() {
            @Override
            public Long apply(String logContent) {
                CollectionTaskLog taskLog = getTaskLog(runTime, null, -1, TaskStatusEnum.SUCCESS, logContent);
                Long taskLogId = collectionTaskService.insertRunLog(taskLog, logContent);
                return taskLogId;
            }
        };

        try {
            generate(sourceTables, targetTables, 1L, null, -1, insertRunLog, null,1L);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    @Override
    public void register() {
        List<TableRegisterEntity> registerList = tableRegisterService.list(new QueryWrapper<TableRegisterEntity>().orderByAsc("batch_num"));
        if(CollectionUtil.isEmpty(registerList))
            return;
        Map<Long, List<TableRegisterEntity>> batchNumMap = registerList.stream().collect(Collectors.groupingBy(TableRegisterEntity::getBatchNum));
        for(Long batchNum : batchNumMap.keySet()){
            if(CollectionUtil.isEmpty(batchNumMap.get(batchNum)))
                continue;
            Map<Long, List<TableRegisterEntity>> runUserMap = batchNumMap.get(batchNum).stream().collect(Collectors.groupingBy(TableRegisterEntity::getRunUser));
            for(Long runUser : runUserMap.keySet()){
                List<TableRegisterEntity> regList = runUserMap.get(runUser);
                if(CollectionUtil.isEmpty(regList)) {
                    continue;
                }
                List<Integer> offLineTableIds=Lists.newArrayList();
                List<Integer> onLineTableIds=Lists.newArrayList();
                List<Integer> deleteOffLineTableIds=Lists.newArrayList();
                List<Integer> deleteOnLineTableIds=Lists.newArrayList();
                for (TableRegisterEntity s : regList) {
                    if (TableRegisterTypeEnum.ONLINE_DELETE.getCode().equals(s.getRegisterType())) {
                        deleteOnLineTableIds.add(s.getTableInfoId());
                    }else  if (TableRegisterTypeEnum.OFFLINE_DELETE.getCode().equals(s.getRegisterType())) {
                        deleteOffLineTableIds.add(s.getTableInfoId());
                    }else  if (s.getRegisterType() != null && s.getRegisterType() < 5) {
                        offLineTableIds.add(s.getTableInfoId());
                    } else if(
                            TableRegisterTypeEnum.ONLINE_ADD.getCode().equals(s.getRegisterType()) ||
                                    TableRegisterTypeEnum.ONLINE_UPDATE.getCode().equals(s.getRegisterType())
                    ){
                        onLineTableIds.add(s.getTableInfoId());
                    }

                }
                try {
                    deleteMetaTables(deleteOffLineTableIds,deleteOnLineTableIds);
                    //当前的源数据表
                    List<CollectionTableInfo> sourceTables = transform(offLineTableIds, onLineTableIds);
                    List<CollectionTableInfo> targetTables = getTargetTables(offLineTableIds,onLineTableIds);
                    generate(sourceTables, targetTables, regList.get(0), null, -1,null);
                } catch (Exception e) {
                    log.error("同步注册错误", e);
                }
            }
        }

        try {
            addRegisterHistory(registerList);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void deleteMetaTables(List<Integer> deleteOffLineTableIds, List<Integer> deleteOnLineTableIds) {
        List<Integer> ids=new ArrayList<>();

        if (CollectionUtils.isNotEmpty(deleteOffLineTableIds)) {
            QueryWrapper<CollectionTableInfo> param = new QueryWrapper<CollectionTableInfo>()
                    .eq("delete_flag", 0).eq("table_type", 1)
                    .in("table_info_id", deleteOffLineTableIds);
            List<CollectionTableInfo> list = collectionTableInfoService.list(param);
            if (CollectionUtils.isNotEmpty(list)) {
                ids.addAll(list.stream().map(CollectionTableInfo::getId).collect(Collectors.toList()));
            }
            bloodRelationshipMapper.delete(
                    new QueryWrapper<TableRelationshipEntity>().in("source_table_id",deleteOffLineTableIds));
        }
        if (CollectionUtils.isNotEmpty(deleteOnLineTableIds)) {
            QueryWrapper<CollectionTableInfo> param = new QueryWrapper<CollectionTableInfo>()
                    .eq("delete_flag", 0).eq("table_type", 2)
                    .in("table_info_id", deleteOnLineTableIds);
            List<CollectionTableInfo> list = collectionTableInfoService.list(param);
            if (CollectionUtils.isNotEmpty(list)) {
                ids.addAll(list.stream().map(CollectionTableInfo::getId).collect(Collectors.toList()));
            }
        }
        if (CollectionUtils.isNotEmpty(ids)) {
            collectionTableInfoService.removeByIds(ids);
            bloodRelationshipMapper.delete(
                    new QueryWrapper<TableRelationshipEntity>().in("source_coll_table_id",ids));
            bloodRelationshipMapper.delete(
                    new QueryWrapper<TableRelationshipEntity>().in("source_table_id",ids));
            bloodRelationshipMapper.delete(
                    new QueryWrapper<TableRelationshipEntity>().in("relation_coll_table_id",ids));

        }
    }

    private List<CollectionTableInfo> transform(List<Integer> offLineTableIds, List<Integer> onLineTableIds) {

        List<CollectionTableInfo> sourceTables = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(offLineTableIds)){
            List<TableInfo> sourceTableList = tableInfoService.list(new QueryWrapper<TableInfo>().in("id", offLineTableIds));
            for (TableInfo t : sourceTableList) {
                CollectionTableInfo tableInfo = new CollectionTableInfo();
                tableInfo.setTableName(t.getTableName());
                tableInfo.setTableAlias(t.getTableAlias());
                tableInfo.setDbName(t.getDbName());
                tableInfo.setTableDesc(t.getDesc());
                tableInfo.setRowsNum(t.getRowsNum());
                tableInfo.setTotalSize(t.getTotalSize());
                tableInfo.setTableInfoId(t.getId());
                tableInfo.setTableType(1);//离线数据
                tableInfo.setSecurityDataLevel(t.getSecurityDataLevel());
                tableInfo.setSourceId(0);
                tableInfo.setRegisterFrom(1);//注册来源 1:数仓  2:外部表
                tableInfo.setTaskType(2);
                tableInfo.setTenantId(t.getTenantId());
                if (t.getCreatePer() != null) {
                    tableInfo.setCreateUser(t.getCreatePer().longValue());
                }
                tableInfo.setUpdateUser(t.getUpdatePer());
                tableInfo.setCreateTime(t.getCreateTime());
                tableInfo.setUpdateTime(t.getUpdateTime());
                sourceTables.add(tableInfo);
            }
        }
        if(CollectionUtil.isNotEmpty(onLineTableIds)){
            List<BdpRealTableInfo> sourceTableList=realTableInfoService.list(new QueryWrapper<BdpRealTableInfo>()
                    .in("id", onLineTableIds));
            for (BdpRealTableInfo t : sourceTableList) {
                CollectionTableInfo tableInfo = new CollectionTableInfo();
                tableInfo.setTableName(t.getTableName());
                //tableInfo.setTableAlias(t.getTopicName());
                tableInfo.setDbName(t.getDbName());
                tableInfo.setTableDesc(t.getTableDesc());
                tableInfo.setRowsNum(0L);
                tableInfo.setTotalSize(0.0);
                tableInfo.setTableInfoId(t.getId());
                tableInfo.setTableType(2);//实时数据
                tableInfo.setSecurityDataLevel(t.getSecurityDataLevel());
                tableInfo.setSourceId(t.getResourceId());
                tableInfo.setTaskType(2);
                tableInfo.setTenantId(t.getTenantId());
                tableInfo.setRegisterFrom(1);//注册来源 1:数仓  2:外部表
                tableInfo.setCreateUser(t.getCreateUser());
                tableInfo.setUpdateUser(t.getUpdateUser());
                tableInfo.setCreateTime(t.getCreateTime());
                tableInfo.setUpdateTime(t.getUpdateTime());
                sourceTables.add(tableInfo);
            }
        }
        return sourceTables;
    }
    private List<CollectionTableInfo> getTargetTables(List<Integer> offLineTableIds, List<Integer> tableIdList){
        ArrayList<CollectionTableInfo> list = Lists.newArrayList();
        if (CollectionUtils.isNotEmpty(offLineTableIds)) {
            List<CollectionTableInfo> offTargetTables = collectionTableInfoService.list(
                    new QueryWrapper<CollectionTableInfo>()
                            .eq("delete_flag", 0).eq("table_type",1)
                            .in("table_info_id", offLineTableIds)
            );
            list.addAll(offTargetTables);

        }
        if (CollectionUtils.isNotEmpty(tableIdList)) {
            List<CollectionTableInfo> onTargetTables = collectionTableInfoService.list(
                    new QueryWrapper<CollectionTableInfo>()
                            .eq("delete_flag", 0).eq("table_type",2)
                            .in("table_info_id", tableIdList)
            );
            list.addAll(onTargetTables);
        }
        return list;
    }

    private void addRegisterHistory(List<TableRegisterEntity> registerList){
        List<TableRegisterHistoryEntity> historyEntityList = new ArrayList<>();
        List<Long> idList = new ArrayList<>();
        registerList.stream().forEach(registerEntity -> {
            TableRegisterHistoryEntity historyEntity = new TableRegisterHistoryEntity();
            BeanUtils.copyProperties(registerEntity, historyEntity);
            historyEntityList.add(historyEntity);
            idList.add(registerEntity.getTableRegisterId());
        });

        tableRegisterService.removeByIds(idList);
        tableRegisterHistoryService.saveBatch(historyEntityList);
    }

    private void  generate(List<CollectionTableInfo> sourceTables, List<CollectionTableInfo> targetTables,
                           TableRegisterEntity registerEntity,
                          Integer defaultSystem, Integer defaultModule,
                          List<CollectionSourceDto> sourceList) throws Exception{
        Function<String, Long> insertRunLog = logContent -> {
            String taskName = TableRegisterTypeEnum.getTypeName(registerEntity.getRegisterType());
            CollectionTableInfo tableInfo = null;
            if(CollectionUtil.isNotEmpty(sourceTables))
                tableInfo = sourceTables.get(0);
            else {
                tableInfo = targetTables.get(0);
            }
            CollectionTaskLog taskLog = getTaskLog(taskName, registerEntity.getRunTime(), registerEntity.getRunUser(),
                    registerEntity.getCost(), defaultSystem, defaultModule,
                    TaskStatusEnum.SUCCESS, tableInfo.getSourceId(), tableInfo.getDbName(), sourceList);
            Long taskLogId = collectionTaskService.insertRunLog(taskLog, logContent);
            return taskLogId;
        };
        //授权的角色信息
        Long roleId = 1L;
        if(registerEntity.getRoleId() != null && !Objects.equals(1L,registerEntity.getRoleId())){
            roleId = registerEntity.getRoleId();
        }
        generate(sourceTables, targetTables, registerEntity.getRunUser(), defaultSystem, defaultModule, insertRunLog, null,roleId);
    }

    private void generate(List<CollectionTableInfo> sourceTables, List<CollectionTableInfo> targetTables, Long runUser,
                          Integer defaultSystem, Integer defaultModule, Function<String, Long> insertRunLog,
                          Consumer<CollectionTableInfo> setRelation, Long roleId) throws Exception{
        Function<CollectionTableInfo,CollectionTableInfo> getRowsAndLength = tableInfo -> {
            CollectionTableInfo rowsAdnLength = new CollectionTableInfo();
            rowsAdnLength.setRowsNum(tableInfo.getRowsNum());
            rowsAdnLength.setTotalSize(tableInfo.getTotalSize());
            return rowsAdnLength;
        };
        Consumer<CollectionTableInfo> setTableInfo = tableInfo -> {
            if (tableInfo.getRowsNum() == null)
                tableInfo.setRowsNum(0L);
            if (tableInfo.getTotalSize() == null)
                tableInfo.setTotalSize(0.0);
            tableInfo.setViewCount(0);
            tableInfo.setTaskType(2);//录入方式 0 手动采集 1自动采集 2接口
            tableInfo.setTempFlag(0);//是否临时表 1 是,0 否
            tableInfo.setCreateTime(new Date());
            if(tableInfo.getCreateUser() == null)
                tableInfo.setCreateUser(runUser);
            tableInfo.setUpdateTime(tableInfo.getCreateTime());
            tableInfo.setUpdateUser(tableInfo.getCreateUser());
            tableInfo.setDeleteFlag(0);//删除 1 是,0 否
            if(StringUtils.isEmpty(tableInfo.getDbName()))
                tableInfo.setDbName(sourceTables.get(0).getDbName());
            tableInfo.setOwnModule(defaultModule);
            if(StringUtils.isEmpty(tableInfo.getSouceConf()))
                tableInfo.setSouceConf("");
            if(tableInfo.getSourceId() == null)
                tableInfo.setSourceId(0);//默认数据源
        };
        Function<CollectionTableInfo,List<CollectionTableField>> getSourceColumns = this::getSourceColumns;
        Function<Integer,List<CollectionTableField>> getTargetTableField = this::getTargetTableField;
        collectionGenerateService.generate(sourceTables, targetTables, roleId, runUser,0L,
                insertRunLog, getRowsAndLength, setTableInfo, getSourceColumns, getTargetTableField, null);
    }


    @Autowired
    private BdpRealTableFieldMapper bdpRealTableFieldMapper;


    private List<CollectionTableField> getSourceColumns(CollectionTableInfo tableInfo) {
        List<CollectionTableField> tableFields = new ArrayList<>();
        if (StoreTableType.OFFLINE.getCode().equals(tableInfo.getTableType())) {
            List<TableFieldInfo> fieldList = tableFieldInfoService.list(new QueryWrapper<TableFieldInfo>()
                    .eq("table_id", tableInfo.getTableInfoId()));
            if (CollectionUtil.isNotEmpty(fieldList))
            {
                for (TableFieldInfo f : fieldList) {
                    CollectionTableField tableField = new CollectionTableField();
                    tableField.setTableId(tableInfo.getId());
                    tableField.setFieldName(f.getFieldName());
                    tableField.setFieldType(f.getFieldType());
                    tableField.setFieldDesc(f.getFieldAlias());
                    tableField.setIsPartition(f.getIsPartition());
                    tableField.setPrimaryKey(f.getPrimaryKey());
                    tableField.setFieldOrder(f.getFieldOrder());
                    tableField.setCreateTime(f.getCreateTime());
                    tableField.setCreateUser(f.getCreatePer() !=null ? f.getCreatePer().longValue() : 0);
                    tableField.setUpdateTime(tableField.getCreateTime());
                    tableField.setUpdateUser(tableField.getUpdateUser());
                    tableField.setDeleteFlag(0);
                    tableField.setTableFieldId(f.getId());
                    tableField.setSecurityDataLevel(f.getSecurityDataLevel());
                    tableFields.add(tableField);
                }
            }
        } else if (StoreTableType.ONLINE.getCode().equals(tableInfo.getTableType())) {
            List<BdpRealTableField> fieldList = bdpRealTableFieldMapper.selectList(
                    new QueryWrapper<BdpRealTableField>().eq("table_id", tableInfo.getTableInfoId())
                            .eq("delete_flag",0));
            if (CollectionUtil.isNotEmpty(fieldList)){
                for (BdpRealTableField f : fieldList) {
                    CollectionTableField tableField = new CollectionTableField();
                    tableField.setTableId(tableInfo.getId());
                    tableField.setFieldName(f.getFieldName());
                    tableField.setFieldType(f.getFieldType());
                    tableField.setFieldDesc(f.getFieldAlias());
                    tableField.setIsPartition(0);
                    tableField.setPrimaryKey(0);
                    tableField.setFieldOrder(f.getFieldOrder());
                    tableField.setCreateTime(f.getCreateTime());
                    tableField.setCreateUser(f.getCreateUser() !=null ? f.getCreateUser() : 0);
                    tableField.setUpdateTime(tableField.getCreateTime());
                    tableField.setUpdateUser(tableField.getUpdateUser());
                    tableField.setDeleteFlag(0);
                    tableField.setTableFieldId(f.getId());
                    tableField.setSecurityDataLevel(f.getSecurityDataLevel());
                    tableFields.add(tableField);
                }
            }
        }
        return tableFields;
    }

    private List<CollectionTableField> getTargetTableField(Integer tableId) {
        return collectionTableField2Service.list(
                new QueryWrapper<CollectionTableField>().eq("table_id", tableId)
                        .eq("delete_flag", 0)
                        .orderByDesc("field_name")
        );
    }

    private CollectionTaskLog getTaskLog(String taskName, Date runTime, Long runUser, long cost,
                                         Integer defaultSystem, Integer defaultModule,
                                         TaskStatusEnum taskStatus,
                                         Integer sourceId, String dbName,
                                         List<CollectionSourceDto> sourceList){
        CollectionTaskLog taskLog = new CollectionTaskLog();
        taskLog.setTaskId(0L);
        taskLog.setTaskName(taskName);
        taskLog.preSave(runUser);
        taskLog.setTaskType(TaskTypeEnum.Interface.getValue());
        taskLog.setConsumerSeconds(cost);
        taskLog.setSourceType("");
        if(sourceId != null)
            taskLog.setSourceId(Long.valueOf(sourceId+""));
        taskLog.setDbName(dbName);
        taskLog.setOwnSystem(defaultSystem);
        taskLog.setOwnModule(defaultModule);
        taskLog.setTaskLeader(runUser);
        taskLog.setTaskRunLeader(runUser);
        taskLog.setRunTime(runTime);
        taskLog.setRunStatus(taskStatus.getValue());
        taskLog.setRunEnable(1);
        taskLog.setSourceId(0L);
        taskLog.setCreateTime(new Date());
        taskLog.setCreateUser(runUser);
        taskLog.setDeleteFlag(0);
        if(CollectionUtil.isNotEmpty(sourceList)){
            StringBuilder sourceStr = new StringBuilder();
            for(CollectionSourceDto sourceDto : sourceList){
                sourceStr.append(sourceDto.getSourceId()).append("@:").append(sourceDto.getDbName()).append(";");
            }
            taskLog.setTaskConfig(sourceStr.substring(0, sourceStr.length() - 1));
        }
        return taskLog;
    }

    private CollectionTaskLog getTaskLog(Date runTime, Integer defaultSystem, Integer defaultModule, TaskStatusEnum taskStatus, String logContent){
        Date endTime = new Date();
        long cost = endTime.getTime() - runTime.getTime();
        CollectionTaskLog taskLog = new CollectionTaskLog();
        taskLog.setTaskId(0L);
        taskLog.setTaskName("初始化数据仓库");
        taskLog.setTaskType(TaskTypeEnum.Interface.getValue());
        taskLog.setConsumerSeconds(cost);
        taskLog.setSourceType("");
        taskLog.setSourceId(0L);
        taskLog.setOwnSystem(defaultSystem);
        taskLog.setOwnModule(defaultModule);
        taskLog.setTaskLeader(1L);
        taskLog.setTaskRunLeader(1L);
        taskLog.setRunStatus(taskStatus.getValue());
        taskLog.setRunEnable(1);
        taskLog.setSourceId(0L);
        taskLog.setRunTime(runTime);
        taskLog.setCreateTime(new Date());
        taskLog.setCreateUser(1L);
        taskLog.setUpdateTime(taskLog.getCreateTime());
        taskLog.setUpdateUser(taskLog.getCreateUser());
        taskLog.setDeleteFlag(0);
        return taskLog;
    }
}
